package com.example.demo.config;

import io.moquette.broker.Server;
import io.moquette.broker.config.IConfig;
import io.moquette.broker.config.MemoryConfig;
import org.eclipse.paho.client.mqttv3.*;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class MqttBroker {

    private Server mqttBroker;
    public static MqttClient mqttClient;
    @Value("${mymqtt.tcp}")
    private String tcpport;
    @Value("${mymqtt.wes}")
    private String wesport;

    private static boolean isConnected = false;
    private static final long reconnectInterval = 5000;
    @PostConstruct
    public void startBroker() throws IOException {
        //启动 Broker
        mqttBroker = new Server();
        Properties configProps = new Properties();
        configProps.setProperty("port", tcpport);
        configProps.setProperty("websocket_port", wesport);
        configProps.setProperty("websocket_path", "/mqtt");
        configProps.setProperty("max_connections", "500");
        IConfig config = new MemoryConfig(configProps);
        try {
            mqttBroker.startServer(config);
            System.out.println("MQTT Broker with SSL/TLS started");
        } catch (Exception e) {
            e.printStackTrace();
        }

        startMqttClient();
    }

    private void startMqttClient() {
        try {
            String brokerUrl = "tcp://localhost:"+tcpport;
            String clientId = "mqttClient";
            mqttClient = new MqttClient(brokerUrl, clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);

            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost: " + cause.getMessage());
                    reconnect();
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    System.out.println("Received message: " + new String(message.getPayload()) + " from topic: " + topic);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // Not used for subscribers
                }
            });

            mqttClient.connect(options);
            mqttClient.subscribe("#"); // 订阅所有主题
            System.out.println("MQTT Client subscribed to all topics.");
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    private void reconnect() {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(() -> {
            try {
                if (!mqttClient.isConnected()) {
                    System.out.println("Attempting to reconnect...");
                    MqttConnectOptions options = new MqttConnectOptions();
                    options.setCleanSession(true);
                    mqttClient.connect(options);
                    mqttClient.subscribe("#");
                    System.out.println("Reconnected successfully.");
                }
            } catch (MqttException e) {
                System.out.println("Reconnection failed: " + e.getMessage());
            }
        }, 0, 5, TimeUnit.SECONDS);
    }


    @PreDestroy
    public void stopBroker() {
        if (mqttBroker != null) {
            mqttBroker.stopServer();
        }
        if (mqttClient != null) {
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }
}


